package com.archgeek.bigdata.flink.platform.arch;

import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.types.RowKind;

/**
 * @author pizhihui
 * @date 2024-06-27 16:26
 */
public class RowDataDemo {


    public static void main(String[] args) {

        LogicalType[] typesK = {new IntType(false), new BigIntType(false)};
        String[] namesK = {"key_shopId", "key_orderId"};
        RowDataSerializer rowDataSerializerKey = new RowDataSerializer(RowType.of(typesK, namesK));

        LogicalType[] typeV = {new VarCharType(false, 8), new ArrayType(new IntType())};
        String[] namesV = {"dt", "priceAmount"};
        RowDataSerializer rowDataSerializerValue = new RowDataSerializer(RowType.of(typeV, namesV));

        GenericRowData grd = new GenericRowData(RowKind.INSERT, 2);
        grd.setField(0, 123);
        grd.setField(1, 123);


        GenericRowData.ofKind(RowKind.UPDATE_BEFORE, 123);
        GenericRowData.ofKind(RowKind.UPDATE_AFTER, 456);

        BinaryRowData keyData = rowDataSerializerKey
                .toBinaryRow(GenericRowData.of(1234, 456))
                .copy();


        BinaryRowData valueData = rowDataSerializerValue
                .toBinaryRow(GenericRowData.of("20240312", new int[]{123, 45})).copy();




    }

}


class KeyValue {

    private RowData k;
    private RowData v;

    private long sequenceNumber;

    private ValueKind valueKind;

}


enum ValueKind {
    ADD, DELETE
}

